package com.cloud.async.flow.web.client;


import com.cloud.async.flow.callback.DefaultGroupCallback;
import com.cloud.async.flow.callback.IGroupCallback;
import com.cloud.async.flow.web.tx.TransactionSource;
import com.cloud.async.flow.wrapper.WorkerWrapper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @author chengrui
 */
public abstract class AbstractFlow {

    /**
     * logger
     */
    private final Log logger = LogFactory.getLog(getClass());


    /**
     * 线程池
     */
    protected ExecutorService executorService;

    /**
     * 事务顶级类
     */
    protected TransactionSource transactionSource;


    /**
     * 异步执行,直到所有都完成,或失败后，发起回调
     *
     * @param timeout
     * @param groupCallback
     * @param workerWrapper
     */
    public void beginWorkAsync(long timeout, IGroupCallback groupCallback, WorkerWrapper... workerWrapper) {
        if (groupCallback == null) {
            groupCallback = new DefaultGroupCallback();
        }
        IGroupCallback finalGroupCallback = groupCallback;
        if (this.executorService != null) {
            this.executorService.submit(() -> {
                try {
                    boolean success = beginWork(timeout, workerWrapper);
                    if (success) {
                        finalGroupCallback.success(Arrays.asList(workerWrapper));
                    } else {
                        finalGroupCallback.failure(Arrays.asList(workerWrapper), new TimeoutException());
                    }
                } catch (ExecutionException | InterruptedException e) {
                    e.printStackTrace();
                    finalGroupCallback.failure(Arrays.asList(workerWrapper), e);
                }
            });
        } else {
            throw new NullPointerException();
        }
    }


    /**
     * 如果想自定义线程池，请传pool。不自定义的话，就走默认的COMMON_POOL
     *
     * @param timeout
     * @param workerWrapper
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    protected boolean beginWork(long timeout, WorkerWrapper... workerWrapper) throws ExecutionException, InterruptedException {
        if (workerWrapper == null || workerWrapper.length == 0) {
            return false;
        }
        List<WorkerWrapper> workerWrappers = Arrays.stream(workerWrapper).collect(Collectors.toList());
        return beginWork(timeout, workerWrappers);
    }

    /**
     * 出发点 （不带事务）
     *
     * @param timeout
     * @param workerWrappers
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public boolean beginWork(long timeout, List<WorkerWrapper> workerWrappers) throws ExecutionException, InterruptedException {
        if (workerWrappers == null || workerWrappers.size() == 0) {
            return false;
        }
        //定义一个map，存放所有的wrapper，key为wrapper的唯一id，value是该wrapper，可以从value中获取wrapper的result
        Map<String, WorkerWrapper> forParamUseWrappers = new ConcurrentHashMap<>();
        CompletableFuture[] futures = new CompletableFuture[workerWrappers.size()];
        for (int i = 0; i < workerWrappers.size(); i++) {
            WorkerWrapper wrapper = workerWrappers.get(i);
            futures[i] = CompletableFuture.runAsync(() -> wrapper.work(this.executorService, timeout, forParamUseWrappers), this.executorService);
        }
        try {
            CompletableFuture.allOf(futures).get(timeout, TimeUnit.MILLISECONDS);
            return true;
        } catch (TimeoutException e) {
            Set<WorkerWrapper> set = new HashSet<>();
            totalWorkers(workerWrappers, set);
            for (WorkerWrapper wrapper : set) {
                wrapper.stopNow();
            }
            return false;
        }
    }


    /**
     * 出发点 （带事务）
     *
     * @param timeout
     * @param workerWrappers
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public boolean beginTransactionWork(long timeout, List<WorkerWrapper> workerWrappers) {
        if (workerWrappers == null || workerWrappers.size() == 0) {
            return false;
        }
        //主线程监控
        CountDownLatch mainThreadLatch = new CountDownLatch(1);
        //监控子线程
        CountDownLatch threadLatch = new CountDownLatch(workerWrappers.size());
        //子线程查看状态是否需要回滚事务
        TransactionRollBack transactionRollBack = new TransactionRollBack(false);

        Map<String, WorkerWrapper> forParamUseWrappers = new ConcurrentHashMap<>();
        CompletableFuture[] futures = new CompletableFuture[workerWrappers.size()];
        for (int i = 0; i < workerWrappers.size(); i++) {
            WorkerWrapper wrapper = workerWrappers.get(i);
            futures[i] = CompletableFuture.runAsync(() ->
                    wrapper.work(
                            this.executorService,
                            timeout,
                            forParamUseWrappers,
                            this.transactionSource,
                            mainThreadLatch,
                            threadLatch,
                            transactionRollBack
                    ), this.executorService);
        }

        try {
            //判断是否超时(直接回滚数据)
            boolean threadLatchAwait = threadLatch.await(timeout, TimeUnit.MILLISECONDS);
            if (!threadLatchAwait) {
                transactionRollBack.setNeedRollBack(true);
                releaseWorkerWrapper(workerWrappers);
                return false;
            }
        } catch (InterruptedException e) {
            logger.error("Main thread waiting timeout:[{}]" + workerWrappers);
            transactionRollBack.setNeedRollBack(true);
            releaseWorkerWrapper(workerWrappers);
        } finally {
            mainThreadLatch.countDown();
        }

        try {
            CompletableFuture.allOf(futures).get();
            return getResult(workerWrappers);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }


    /**
     * 获取返回值是否正常执行结束
     *
     * @param workerWrappers
     * @return
     */
    private boolean getResult(List<WorkerWrapper> workerWrappers) {
        Boolean ret = true;
        for (WorkerWrapper workerWrapper : workerWrappers) {
            if (workerWrapper.getWorkResult().getEx() != null) {
                ret = false;
                break;
            }
            if (workerWrapper.getNextWrappers() != null) {
                ret = getResult(workerWrapper.getNextWrappers());
                if (!ret) {
                    break;
                }
            }
        }
        return ret;
    }


    /**
     * 总共多少个执行单元
     *
     * @param workerWrappers
     * @param set
     */
    @SuppressWarnings("unchecked")
    private void totalWorkers(List<WorkerWrapper> workerWrappers, Set<WorkerWrapper> set) {
        set.addAll(workerWrappers);
        for (WorkerWrapper wrapper : workerWrappers) {
            if (wrapper.getNextWrappers() == null) {
                continue;
            }
            List<WorkerWrapper> wrappers = wrapper.getNextWrappers();
            totalWorkers(wrappers, set);
        }
    }

    /**
     * END
     *
     * @param workerWrappers
     */
    private void releaseWorkerWrapper(List<WorkerWrapper> workerWrappers) {
        Set<WorkerWrapper> set = new HashSet<>();
        totalWorkers(workerWrappers, set);
        for (WorkerWrapper wrapper : set) {
            wrapper.stopNow();
        }
    }
}
